
package com.huaweicloud.saaslanding.install.applicationservice.impl;

import com.huaweicloud.saaslanding.basic.database.mybatis.conditions.query.QueryWrap;
import com.huaweicloud.saaslanding.dao.common.entity.tenant.Tenant;
import com.huaweicloud.saaslanding.dao.common.repository.tenant.TenantBaseMapper;
import com.huaweicloud.saaslanding.install.applicationservice.KafkaService;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ListTopicsOptions;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;

/**
 * 功能描述
 *
 * @since 2022-12-26
 */
@Slf4j
@Service
@ConditionalOnExpression("${saaslanding-install.lts.enabled:false} && 'Kafka'.equalsIgnoreCase('${saaslanding-install.lts.consumer}')")
public class KafkaServiceImpl implements KafkaService {
    @Resource
    private TenantBaseMapper tenantBaseMapper;

    @Resource
    private AdminClient adminClient;

    /**
     * 删除Kafka主题
     */
    @Override
    public void deleteKafka() {
        try {
            // 主题名称是租户code，删除租户时主题可能没有删除主题，查询所有租户
            List<String> tenantList = tenantBaseMapper.selectList(new QueryWrap<Tenant>().select("code"))
                .stream()
                .map(Tenant::getCode)
                .collect(Collectors.toList());

            Set<String> topicSet = adminClient.listTopics(new ListTopicsOptions().listInternal(false)).names().get();
            tenantList = tenantList.stream().filter(topicSet::contains).collect(Collectors.toList());

            // 不需手动删除消费组，删除主题的同时会自动删除对应的消费组
            adminClient.deleteTopics(tenantList);
        } catch (ExecutionException | InterruptedException e) {
            log.warn("删除Kafka主题失败，可能是连接信息出错！！！");
            log.info("请手动前往Kafka删除主题.");
        }
    }
}
